package com.summer.rx.impl;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * 被观察者
 */
public abstract class Observed<T> {

    private static final Handler handler = new Handler();

    public abstract void subscribe(Observer<T> observer);

    public static <T> Observed<T> create(Observed<T> observed) {
        return observed;
    }

    public static <T> Observed<T> just(T... data) {
        return new Observed<T>() {
            @Override
            public void subscribe(Observer<T> observer) {
                if(data != null) {
                    for(T value : data) {
                        observer.onNext(value);
                    }
                }
            }
        };
    }

    public Observed<T> flatMap() {
        return new Observed<T>() {
            @Override
            public void subscribe(final Observer<T> observer) {
                Observer<T> newObserver = new Observer<T>() {
                    @Override
                    public void onNext(T value) {
                        observer.onNext(value);
                    }

                    @Override
                    public void onComplete() {
                        observer.onComplete();
                    }

                    @Override
                    public void onError(Throwable ex) {
                        observer.onError(ex);
                    }
                };
                Observed.this.<T>subscribe(newObserver);
            }
        };
    }

    public <R> Observed<R> map(final Function<T, R> function) {
        return new Observed<R>() {
            @Override
            public void subscribe(Observer<R> observer) {
                Observer<T> newObserver = new Observer<T>() {
                    @Override
                    public void onNext(T value) {
                        observer.onNext(function.apply(value));
                    }

                    @Override
                    public void onComplete() {
                        observer.onComplete();
                    }

                    @Override
                    public void onError(Throwable ex) {
                        observer.onError(ex);
                    }
                };
                Observed.this.<T>subscribe(newObserver);
            }
        };
    }

    public Observed<T> subscribeOn() {
        return new Observed<T>() {
            @Override
            public void subscribe(final Observer<T> observer) {
                Observer<T> newObserver = new Observer<T>() {
                    @Override
                    public void onNext(T value) {
                        new Thread(() -> observer.onNext(value)).start();
                    }

                    @Override
                    public void onComplete() {
                        observer.onComplete();
                    }

                    @Override
                    public void onError(Throwable ex) {
                        observer.onError(ex);
                    }
                };
                Observed.this.<T>subscribe(newObserver);
            }
        };
    }

    public Observed<T> subscribeBy() {
        return new Observed<T>() {
            @Override
            public void subscribe(final Observer<T> observer) {
                Observer<T> newObserver = new Observer<T>() {
                    @Override
                    public void onNext(T value) {
                        handler.post(() -> observer.onNext(value));
                    }

                    @Override
                    public void onComplete() {
                        observer.onComplete();
                    }

                    @Override
                    public void onError(Throwable ex) {
                        observer.onError(ex);
                    }
                };
                Observed.this.<T>subscribe(newObserver);
            }
        };
    }

    @SuppressWarnings("unused")
	public void accept() {
        Observed.this.<T>subscribe(new Observer<T>() {
            @Override
            public void onNext(T value) {
            	System.out.println("end");
            }

            @Override
            public void onComplete() {

            }

            @Override
            public void onError(Throwable ex) {

            }
        });
    }

    private static class Handler implements Runnable {
    	private DelayQueue<DelaySendTask> queue = new DelayQueue<>();
    	
    	public Handler() {
    		new Thread(this, "Handler").start();
    	}
    	
    	public void post(Runnable run) {
    		post(run, 0);
    	}
    	
    	public void post(Runnable run, int delayTime) {
    		queue.add(new DelaySendTask(run, delayTime));
    	}

		@Override
		public void run() {
			while (true) {
	            try {
	                DelaySendTask take = queue.take();
	                if (take != null) {
	                    take.execTask();
	                }
	            } catch (InterruptedException e) {
	                e.printStackTrace();
	            }
	        }
		}
    	
		private class DelaySendTask implements Delayed {
	        private final long availableTime;
	        private final Runnable runnable;

	        public DelaySendTask(Runnable runnable, int delayTime) {
	            this.runnable = runnable;
	            this.availableTime = delayTime + System.currentTimeMillis();
	        }

	        public void execTask() {
	            if(runnable != null) {
	            	runnable.run();
	            }
	        }

	        @Override
	        public long getDelay(TimeUnit timeUnit) {
	            //判断 availableTime 是否大于当前系统时间，并将结果转换成 MILLISECONDS
	            long diffTime = availableTime - System.currentTimeMillis();
	            return timeUnit.convert(diffTime, TimeUnit.MILLISECONDS);
	        }

	        @Override
	        public int compareTo(Delayed delayed) {
	            // compareTo 用在 DelaySendTask 的排序
	            return (int) (this.availableTime - ((DelaySendTask) delayed).availableTime);
	        }
	    }
    }
    
}
